package com.shujia.flink.source

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction, RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala._

object Demo3MysqlSource {
  def main(args: Array[String]): Unit = {


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    env.setParallelism(2)

    //使用mysql source

    val studentDS: DataStream[Student] = env.addSource(new MysqlSource)


    val clazzNum: DataStream[(String, Int)] = studentDS.map(stu => (stu.clazz, 1)).keyBy(_._1).sum(1)

    clazzNum.print()


    env.execute()

  }

}

case class Student(id: Int, name: String, age: Int, gender: String, clazz: String)


/**
  * 自定义source的接口
  *
  * SourceFunction   单线程source
  * RichSourceFunction  单线程source    , 多了一个open 和close 方法用于打开和关闭链接
  * ParallelSourceFunction 并行的source  可以提高读取数据的效率
  * RichParallelSourceFunction 并行的source 多了一个open 和close 方法用于打开和关闭链接
  *
  */

/**
  * 自定义source 读取mysql   是一个有界流
  *
  */
class MysqlSource extends RichSourceFunction[Student] {

  var con: Connection = _

  /**
    * 打开网络链接，只运行一次,  在run 方法之前执行
    *
    */
  override def open(parameters: Configuration): Unit = {
    Class.forName("com.mysql.jdbc.Driver")
    con = DriverManager.getConnection("jdbc:mysql://master:3306/student", "root", "123456")
  }


  override def run(ctx: SourceFunction.SourceContext[Student]): Unit = {

    /**
      * 读取mysql中的数据
      *
      */


    val stat: PreparedStatement = con.prepareStatement("select * from student")
    val resultSet: ResultSet = stat.executeQuery()

    while (resultSet.next()) {

      val id: Int = resultSet.getInt("id")
      val name: String = resultSet.getString("name")
      val age: Int = resultSet.getInt("age")
      val gender: String = resultSet.getString("gender")
      val clazz: String = resultSet.getString("clazz")


      val student = Student(id, name, age, gender, clazz)


      /**
        * 将数据发送到下游
        *
        */
      ctx.collect(student)

    }
  }

  /**
    * close 在run 之后执行 一般用户关闭链接
    *
    */

  override def close(): Unit = {
    con.close()
  }


  override def cancel(): Unit = {

  }

}